package com.shujia.flink.source;

import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;


public class Demo04MySource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.addSource(new MySource()).print();

        env.execute();

    }
}

class MySource implements SourceFunction<String> {

    /**
     * run方法只会在程序运行时执行一次
     */
    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        System.out.println("运行了run方法");
        while (true) {
            ctx.collect("向下游发送了一条数据");
            Thread.sleep(RandomUtils.nextInt(1, 4) * 1000L);
        }
    }

    @Override
    public void cancel() {
        System.out.println("任务取消了");
    }
}